package com.learn.lb.storm.wordcount;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;

import java.util.Map;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * 数据源
 * @author laibo
 * @since 2019/9/18 09:24
 */
public class WcSpout extends BaseRichSpout {

    private static final String[] TEST_SOURCE = {"a", "b", "c", "d", "e", "f"};

    private SpoutOutputCollector collector;

    /**
     * 定义发射的字段类型，是第一个要执行的方法
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("wcField"));
    }


    /**
     * 打开stream流资源，只会执行一次
     */
    @Override
    public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
        this.collector = spoutOutputCollector;
    }

    /**
     * 循环执行，向外发送 Tuple
     * 这是Spout最主要的方法，在这里我们读取文本文件，并把它的每一行发射出去（给bolt）
     * 这个方法会不断被调用，为了降低它对CPU的消耗，当任务完成时让它sleep一下
     * nextTuple会在同一个循环内被ack()和fail()周期性的调用。没有任务时它必须释放对线程的控制，
     * 其它方法才有机会得以执行。因此nextTuple的第一行就要检查是否已处理完成。
     * 如果完成了，为了降低处理器负载，会在返回前休眠一秒。如果任务完成了，文件中的每一行都已被读出并分发了
     */
    @Override
    public void nextTuple() {
        //通过emit发送tuple
        this.collector.emit(new Values(TEST_SOURCE[new Random().nextInt(TEST_SOURCE.length)]));
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
